package jsu.weizixuan.likesystem.listener;

import cn.hutool.core.lang.Pair;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import jsu.weizixuan.likesystem.constant.MQConstant;
import jsu.weizixuan.likesystem.job.TransactionalMethod;
import jsu.weizixuan.likesystem.model.msg.ThumbEvent;
import jsu.weizixuan.likesystem.util.RedisKeyUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @Title: ThumbConsumer
 * @Author 魏子轩
 * @Package jsu.weizixuan.likesystem.listener
 * @Date 2025/9/27 4:29
 * @description: 点赞业务消费者
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class ThumbConsumer {

    private final TransactionalMethod transactionalMethod;
    private final RedisTemplate<String, Object> redisTemplate;
    private final RabbitTemplate rabbitTemplate;

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = MQConstant.THUMB_QUEUE, durable = "true",
                    arguments = {
                            @Argument(name = "x-dead-letter-exchange", value = MQConstant.DLX_THUMB_EXCHANGE),
                            @Argument(name = "x-dead-letter-routing-key", value = MQConstant.DLX_THUMB_ROUTING_KEY)
                    }),
            exchange = @Exchange(name = MQConstant.THUMB_EXCHANGE),
            key = {MQConstant.THUMB_ROUTING_KEY}
    ), containerFactory = "batchContainerFactory")
    public void processBatch(List<Message> messages, Channel channel) {

        // 批量处理消息,将消息体转为ThumbEvent，并将消息的deliveryTag保存在ThumbEvent中
        List<ThumbEvent> thumbEventList = messages.stream().map(message -> {
//            try {
//                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//            } catch (IOException e) {
//                throw new RuntimeException(e);
//            }
            ThumbEvent thumbEvent = JSON.parseObject(new String(message.getBody()), ThumbEvent.class);
            if (thumbEvent.getMessage() != null) {
                Integer retrycount = (Integer) thumbEvent.getMessage().getMessageProperties().getHeaders().get("x-retry-count");
                message.getMessageProperties().getHeaders().put("x-retry-count",retrycount+1);
            }
            thumbEvent.setMessage(message);
            return thumbEvent;
        }).toList();

        // 处理内容缺失的消息
        thumbEventList.stream().filter(Objects::isNull).forEach(thumbEvent -> {
            log.error("消息内容缺失：{}", thumbEvent);
            try {
                channel.basicAck(thumbEvent.getMessage().getMessageProperties().getDeliveryTag(), false);
            } catch (IOException e) {
                log.error("(内容缺失)消息确认失败：{}", thumbEvent);
                throw new RuntimeException(e);
            }
        });

        // 按(userId, blogId)分组，并获取每个分组的最新事件
        Map<Pair<Long, Long>, ThumbEvent> thumbEventMap = thumbEventList.stream().filter(Objects::nonNull).collect(Collectors.groupingBy(
                e -> Pair.of(e.getUserId(), e.getBlogId()),
                Collectors.collectingAndThen(
                        Collectors.toList(),
                        list -> {
                            // 按时间升序排序，取最后一个作为最新事件
                            list.sort(Comparator.comparing(ThumbEvent::getEventTime));
                            if (list.size() % 2 == 0) {
                                return null;
                            }
                            return list.get(list.size() - 1);
                        }
                )
        ));

        thumbEventMap.forEach((userBlogPair, event) -> {
            if (event == null) {
                return;
            }
            ThumbEvent.EventType finalAction = event.getType();
            try {
                if (finalAction == ThumbEvent.EventType.INCR) {
                    transactionalMethod.doThumbMethod(event);
                } else {
                    transactionalMethod.undoThumbMethod(event);
                }
                channel.basicAck(event.getMessage().getMessageProperties().getDeliveryTag(), false);
            } catch (Exception e) {
                log.error("消息消费失败,开始重试或入库人工介入 message -> {}, errorMsg -> {}", event, e.getMessage());
                Integer retryCount = (Integer) event.getMessage().getMessageProperties().getHeaders().getOrDefault("x-retry-count", 0);
                event.getMessage().getMessageProperties().getHeaders().put("x-retry-count", retryCount);
                if (retryCount < 3) {
                    log.error("消息消费失败,开始第 {} 次重试 message -> {}", retryCount+1, event);
                    try {
                        channel.basicAck(event.getMessage().getMessageProperties().getDeliveryTag(), false);
                    } catch (IOException ex) {
                        throw new RuntimeException(ex);
                    }
                    rabbitTemplate.convertAndSend(MQConstant.THUMB_EXCHANGE, MQConstant.THUMB_ROUTING_KEY, JSON.toJSONString(event));
                } else {
                    log.error("消息重试次数超限，进入死信队列中进行处理 message -> {}", event);
                    try {
                        channel.basicNack(event.getMessage().getMessageProperties().getDeliveryTag(), false, false);
                    } catch (IOException ex) {
                        log.error("消息转发到死信队列失败： {}", e.getMessage());
                    }
                }
            }
        });
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = MQConstant.DLX_THUMB_QUEUE, durable = "true"),
            exchange = @Exchange(name = MQConstant.DLX_THUMB_EXCHANGE)
    ))
    public void comsumerDlq(List<Message> messages, Channel channel) {
        messages.forEach(message -> {
            log.info("进入死信队列消息：{}", message);
            try {
                ThumbEvent thumbEvent = JSON.parseObject(new String(message.getBody()), ThumbEvent.class);
                if (thumbEvent != null) {
                    redisTemplate.opsForHash().delete(RedisKeyUtil.getUserThumbKey(thumbEvent.getUserId()), thumbEvent.getBlogId().toString());
                }
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                log.info("死信队列消息处理成功");
            } catch (IOException e) {
                log.error("处理死信消息时发生异常: ", e);
                try {
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                } catch (IOException ex) {
                    log.error("死信队列消息丢弃失败: ", ex);
                    throw new RuntimeException(ex);
                }
            }
        });
    }
}
